//package canal;
//
//import com.alibaba.otter.canal.protocol.position.EntryPosition;
//import com.alibaba.otter.canal.protocol.position.LogPosition;
//
//import java.io.IOException;
//import java.io.Serializable;
//import java.io.UnsupportedEncodingException;
//import java.net.InetSocketAddress;
//import java.nio.ByteBuffer;
//import java.nio.channels.SocketChannel;
//import java.nio.charset.Charset;
//import java.util.List;
//
///**
// * @Author: admin
// * @Description:
// * @Date: 2018/11/19 15:48
// */
//
//
//public class Slave
//{
////
////    // 连接所有的scramble_buff
////    private static byte[] joinAndCreateScrambleBuff(HandshakeInitializationPacket handshakePacket) throws IOException
////    {
////        byte[] dest = new byte[handshakePacket.seed.length + handshakePacket.restOfScrambleBuff.length];
////        System.arraycopy(handshakePacket.seed, 0, dest, 0, handshakePacket.seed.length);
////        System.arraycopy(handshakePacket.restOfScrambleBuff, 0, dest, handshakePacket.seed.length,
////                handshakePacket.restOfScrambleBuff.length);
////        return dest;
////    }
////
////    private static void auth323(SocketChannel channel, byte packetSequenceNumber, byte[] seed, String password)
////            throws IOException
////    {
////        // auth 323
////        Reply323Packet r323 = new Reply323Packet();
////        // 1.对密码进行加密
////        if (password != null && password.length() > 0)
////        {
////            r323.seed = MySQLPasswordEncrypter.scramble323(password, new String(seed)).getBytes();
////        }
////        byte[] b323Body = r323.toBytes();
////
////        HeaderPacket h323 = new HeaderPacket();
////        h323.setPacketBodyLength(b323Body.length);
////        h323.setPacketSequenceNumber((byte) (packetSequenceNumber + 1));
////
////        PacketManager.write(channel, new ByteBuffer[] { ByteBuffer.wrap(h323.toBytes()), ByteBuffer.wrap(b323Body) });
////        System.out.println("client 323 authentication packet is sent out.");
////        // check auth result
////        HeaderPacket header = PacketManager.readHeader(channel, 4);
////        byte[] body = PacketManager.readBytes(channel, header.getPacketBodyLength());
////        assert body != null;
////        switch (body[0])
////        {
////            case 0:
////                break;
////            case -1:
////                ErrorPacket err = new ErrorPacket();
////                err.fromBytes(body);
////                throw new IOException("Error When doing Client Authentication:" + err.toString());
////            default:
////                throw new IOException("unpexpected packet with field_count=" + body[0]);
////        }
////    }
////
////    // 更新操作
////    public static OKPacket update(SocketChannel channel, String updateString) throws IOException
////    {
////        QueryCommandPacket cmd = new QueryCommandPacket();
////        cmd.setQueryString(updateString);
////        byte[] bodyBytes = cmd.toBytes();
////        PacketManager.write(channel, bodyBytes);
////
////        System.out.println("read update result...");
////        byte[] body = PacketManager.readBytes(channel, PacketManager.readHeader(channel, 4).getPacketBodyLength());
////        if (body[0] < 0)
////        {
////            ErrorPacket packet = new ErrorPacket();
////            packet.fromBytes(body);
////            throw new IOException(packet + "\n with command: " + updateString);
////        }
////
////        OKPacket packet = new OKPacket();
////        packet.fromBytes(body);
////        return packet;
////    }
////
////    // 查询操作
////    public static ResultSetPacket query(SocketChannel channel, String queryString) throws IOException
////    {
////        QueryCommandPacket cmd = new QueryCommandPacket();
////        cmd.setQueryString(queryString);
////        byte[] bodyBytes = cmd.toBytes();
////        PacketManager.write(channel, bodyBytes);
////        byte[] body = readNextPacket(channel);
////
////        if (body[0] < 0)
////        {
////            ErrorPacket packet = new ErrorPacket();
////            packet.fromBytes(body);
////            throw new IOException(packet + "\n with command: " + queryString);
////        }
////
////        ResultSetHeaderPacket rsHeader = new ResultSetHeaderPacket();
////        rsHeader.fromBytes(body);
////
////        List<FieldPacket> fields = new ArrayList<FieldPacket>();
////        for (int i = 0; i < rsHeader.getColumnCount(); i++)
////        {
////            FieldPacket fp = new FieldPacket();
////            fp.fromBytes(readNextPacket(channel));
////            fields.add(fp);
////        }
////
////        readEofPacket(channel);
////
////        List<RowDataPacket> rowData = new ArrayList<RowDataPacket>();
////        while (true)
////        {
////            body = readNextPacket(channel);
////            if (body[0] == -2)
////            {
////                break;
////            }
////            RowDataPacket rowDataPacket = new RowDataPacket();
////            rowDataPacket.fromBytes(body);
////            rowData.add(rowDataPacket);
////        }
////
////        ResultSetPacket resultSet = new ResultSetPacket();
////        resultSet.getFieldDescriptors().addAll(fields);
////        for (RowDataPacket r : rowData)
////        {
////            resultSet.getFieldValues().addAll(r.getColumns());
////        }
////        resultSet.setSourceAddress(channel.socket().getRemoteSocketAddress());
////
////        return resultSet;
////    }
////
////    private static void readEofPacket(SocketChannel channel) throws IOException
////    {
////        byte[] eofBody = readNextPacket(channel);
////        if (eofBody[0] != -2)
////        {
////            throw new IOException("EOF Packet is expected, but packet with field_count=" + eofBody[0] + " is found.");
////        }
////    }
////
////    protected static byte[] readNextPacket(SocketChannel channel) throws IOException
////    {
////        HeaderPacket h = PacketManager.readHeader(channel, 4);
////        return PacketManager.readBytes(channel, h.getPacketBodyLength());
////    }
//
//    public static void main(String[] args) throws Exception
//    {
//        // MysqlConnector.connect()
//        // 1.连接到MysqlServer，同时设置一些参数
//        int soTimeout = 30 * 1000;
//        int receiveBufferSize = 16 * 1024;
//        int sendBufferSize = 16 * 1024;
//        String ip = "127.0.0.1";
//        int port = 3306;
//        String username = "root";
//        String password = "root";
//
//        SocketChannel channel = SocketChannel.open();
//        InetSocketAddress masterAddress = new InetSocketAddress(ip, port);
//
//        channel.socket().setKeepAlive(true);
//        channel.socket().setReuseAddress(true);
//        channel.socket().setSoTimeout(soTimeout);
//        channel.socket().setTcpNoDelay(true);
//        channel.socket().setReceiveBufferSize(receiveBufferSize);
//        channel.socket().setSendBufferSize(sendBufferSize);
//
//        channel.connect(masterAddress);
//
//        // MysqlConnector.negotiate()
//        // 2.读取消息头信息
//        // 消息头，头3个byte是消息长度，1个byte是序列号
//        HeaderPacket header = PacketManager.readHeader(channel, 4);
//        System.out.println("header length=" + header.getPacketBodyLength());
//        // 3.读取指定长度的报文
//        byte[] body = PacketManager.readBytes(channel, header.getPacketBodyLength());
//        // 4.错误检查
//        if (body[0] < 0)
//        {// check field_count
//            if (body[0] == -1)
//            {
//                ErrorPacket error = new ErrorPacket();
//                error.fromBytes(body);
//                throw new IOException("handshake exception:\n" + error.toString());
//            }
//            else if (body[0] == -2)
//            {
//                throw new IOException("Unexpected EOF packet at handshake phase.");
//            }
//            else
//            {
//                throw new IOException("unpexpected packet with field_count=" + body[0]);
//            }
//        }
//        // 5.握手消息初始化
//        HandshakeInitializationPacket handshakePacket = new HandshakeInitializationPacket();
//        handshakePacket.fromBytes(body);// 解析握手消息
//
//        long connectionId = -1;
//        byte charsetNumber = 33;
//        String defaultSchema = "testdb";
//        connectionId = handshakePacket.threadId; // 记录一下connection
//
//        System.out
//                .println("handshake initialization packet received, prepare the client authentication packet to send");
//        // 6.组织认证消息
//        ClientAuthenticationPacket clientAuth = new ClientAuthenticationPacket();
//        clientAuth.setCharsetNumber(charsetNumber);
//
//        clientAuth.setUsername(username);// 用户名
//        clientAuth.setPassword(password);// 密码
//        clientAuth.setServerCapabilities(handshakePacket.serverCapabilities);// 容错情况，这个消息是握手消息中拿到的
//        clientAuth.setDatabaseName(defaultSchema);// 默认数据库
//        clientAuth.setScrumbleBuff(joinAndCreateScrambleBuff(handshakePacket));// 将两个不同位置的scramble_buff合并到一个数组中
//
//        byte[] clientAuthPkgBody = clientAuth.toBytes();
//        HeaderPacket h = new HeaderPacket();
//        h.setPacketBodyLength(clientAuthPkgBody.length);// 设置报文内容长度
//        h.setPacketSequenceNumber((byte) (header.getPacketSequenceNumber() + 1));// 序列+1
//
//        // 7.发送认证信息
//        PacketManager.write(channel,
//                new ByteBuffer[] { ByteBuffer.wrap(h.toBytes()), ByteBuffer.wrap(clientAuthPkgBody) });
//        System.out.println("client authentication packet is sent out.");
//
//        // Mysql会立刻返回信息,解析Mysql的第二个信息
//        // check auth result
//        // 8.读取消息头
//        header = null;
//        header = PacketManager.readHeader(channel, 4);
//        body = null;
//        // 9.读取消息体
//        body = PacketManager.readBytes(channel, header.getPacketBodyLength());
//        assert body != null;
//        if (body[0] < 0)
//        {
//            if (body[0] == -1)
//            {
//                ErrorPacket err = new ErrorPacket();
//                err.fromBytes(body);
//                throw new IOException("Error When doing Client Authentication:" + err.toString());
//            }
//            else if (body[0] == -2)
//            {
//                auth323(channel, header.getPacketSequenceNumber(), handshakePacket.seed, password);
//            }
//            else
//            {
//                throw new IOException("unpexpected packet with field_count=" + body[0]);
//            }
//        }
//
//        // Connection 已经建立完成了，接下来就可以去请求当前binlog文件名称，位置等信息。
//
//
//
//
//        // 更新数据库信息
//        try
//        {
//            update(channel, "set wait_timeout=9999999");
//        }
//        catch (Exception e)
//        {
//            e.printStackTrace();
//        }
//        try
//        {
//            update(channel, "set net_write_timeout=1800");
//        }
//        catch (Exception e)
//        {
//            e.printStackTrace();
//        }
//
//        try
//        {
//            update(channel, "set net_read_timeout=1800");
//        }
//        catch (Exception e)
//        {
//        }
//
//        try
//        {
//            // 设置服务端返回结果时不做编码转化，直接按照数据库的二进制编码进行发送，由客户端自己根据需求进行编码转化
//            update(channel, "set names 'binary'");
//        }
//        catch (Exception e)
//        {
//            e.printStackTrace();
//        }
//
//        try
//        {
//            // mysql5.6针对checksum支持需要设置session变量
//            // 如果不设置会出现错误： Slave can not handle replication events with the
//            // checksum that master is configured to log
//            // 但也不能乱设置，需要和mysql server的checksum配置一致，不然RotateLogEvent会出现乱码
//            // '@@global.binlog_checksum'需要去掉单引号,在mysql 5.6.29下导致master退出
//            update(channel, "set @master_binlog_checksum= @@global.binlog_checksum");
//        }
//        catch (Exception e)
//        {
//            e.printStackTrace();
//
//        }
//
//        // 查询binlog_checksum
//        ResultSetPacket rs = null;
//        try
//        {
//            rs = query(channel, "select @master_binlog_checksum");
//        }
//        catch (IOException e)
//        {
//            e.printStackTrace();
//        }
//        int binlogChecksum;
//
//        List<String> columnValues = rs.getFieldValues();
//        if (columnValues != null && columnValues.size() >= 1 && columnValues.get(0).toUpperCase().equals("CRC32"))
//        {
//            binlogChecksum = LogEvent.BINLOG_CHECKSUM_ALG_CRC32;
//        }
//        else
//        {
//            binlogChecksum = LogEvent.BINLOG_CHECKSUM_ALG_OFF;
//        }
//
//        // 10.发送请求查询位置
//        ResultSetPacket packet = query(channel, "show master status");
//        List<String> fields = packet.getFieldValues();
//        if (fields == null || fields.size() == 0)
//        {
//            System.out.println("无法找到当前的位置");
//            return;
//        }
//        EntryPosition endPosition = new EntryPosition(fields.get(0), Long.valueOf(fields.get(1)));
//
//        System.out.println(
//                String.format("fileName= %s   pos=%d", endPosition.getJournalName(), endPosition.getPosition()));
//        // 实际上这个文件名称和位置是需要存储到本地的，然后去进行同步
//
//        // 10.发送申请binlog的命令
//        BinlogDumpCommandPacket binlogDumpCmd = new BinlogDumpCommandPacket();
//        // 这个数据是需要查出来的
//        binlogDumpCmd.binlogFileName = endPosition.getJournalName();
//        binlogDumpCmd.binlogPosition = endPosition.getPosition();
//        binlogDumpCmd.slaveServerId = 3;
//        byte[] cmdBody = binlogDumpCmd.toBytes();
//
//        HeaderPacket binlogDumpHeader = new HeaderPacket();
//        binlogDumpHeader.setPacketBodyLength(cmdBody.length);
//        binlogDumpHeader.setPacketSequenceNumber((byte) 0x00);
//        PacketManager.write(channel,
//                new ByteBuffer[] { ByteBuffer.wrap(binlogDumpHeader.toBytes()), ByteBuffer.wrap(cmdBody) });
//
//        DirectLogFetcher fetcher = new DirectLogFetcher();
//
//        // 11.启动读取channel的线程
//        fetcher.start(channel);
//
//        LogDecoder decoder = new LogDecoder(LogEvent.UNKNOWN_EVENT, LogEvent.ENUM_END_EVENT);
//        LogContext context = new LogContext();
//        context.setLogPosition(new LogPosition(binlogDumpCmd.binlogFileName));
//        context.setFormatDescription(new FormatDescriptionLogEvent(4, binlogChecksum));
//        // 12.循环读取消息
//        while (fetcher.fetch())
//        {
//            LogEvent event = null;
//            event = decoder.decode(fetcher, context);
//
//            if (event == null)
//            {
//                throw new RuntimeException("parse failed");
//            }
//
//            int eventType = event.getHeader().getType();
//            System.out.println("eventType=" + eventType);
//            switch (eventType)
//            {
//                case LogEvent.ROTATE_EVENT:
//                    // binlogFileName = ((RotateLogEvent)
//                    // event).getFilename();
//                    break;
//                case LogEvent.WRITE_ROWS_EVENT_V1:
//                case LogEvent.WRITE_ROWS_EVENT:
//                    parseRowsEvent(endPosition.getJournalName(), (WriteRowsLogEvent) event);
//                    break;
//                case LogEvent.UPDATE_ROWS_EVENT_V1:
//                case LogEvent.UPDATE_ROWS_EVENT:
//                    // parseRowsEvent((UpdateRowsLogEvent) event);
//                    break;
//                case LogEvent.DELETE_ROWS_EVENT_V1:
//                case LogEvent.DELETE_ROWS_EVENT:
//                    // parseRowsEvent((DeleteRowsLogEvent) event);
//                    break;
//                case LogEvent.QUERY_EVENT:
//                    // parseQueryEvent((QueryLogEvent) event);
//                    break;
//                case LogEvent.ROWS_QUERY_LOG_EVENT:
//                    // parseRowsQueryEvent((RowsQueryLogEvent) event);
//                    break;
//                case LogEvent.ANNOTATE_ROWS_EVENT:
//                    break;
//                case LogEvent.XID_EVENT:
//                    break;
//                default:
//                    break;
//            }
//        }
//
//    }
//
//    protected static Charset charset = Charset.forName("utf-8");
//
//    protected static void parseRowsEvent(String binlogFileName, RowsLogEvent event)
//    {
//        try
//        {
//            System.out.println(String.format("================> binlog[%s:%s] , name[%s,%s]", binlogFileName,
//                    event.getHeader().getLogPos() - event.getHeader().getEventLen(), event.getTable().getDbName(),
//                    event.getTable().getTableName()));
//            RowsLogBuffer buffer = event.getRowsBuf(charset.name());
//            BitSet columns = event.getColumns();
//            BitSet changeColumns = event.getChangeColumns();
//            while (buffer.nextOneRow(columns))
//            {
//                // 处理row记录
//                int type = event.getHeader().getType();
//                if (LogEvent.WRITE_ROWS_EVENT_V1 == type || LogEvent.WRITE_ROWS_EVENT == type)
//                {
//                    // insert的记录放在before字段中
//                    parseOneRow(event, buffer, columns, true);
//                }
//                else if (LogEvent.DELETE_ROWS_EVENT_V1 == type || LogEvent.DELETE_ROWS_EVENT == type)
//                {
//                    // delete的记录放在before字段中
//                    parseOneRow(event, buffer, columns, false);
//                }
//                else
//                {
//                    // update需要处理before/after
//                    System.out.println("-------> before");
//                    parseOneRow(event, buffer, columns, false);
//                    if (!buffer.nextOneRow(changeColumns))
//                    {
//                        break;
//                    }
//                    System.out.println("-------> after");
//                    parseOneRow(event, buffer, changeColumns, true);
//                }
//
//            }
//        }
//        catch (Exception e)
//        {
//            throw new RuntimeException("parse row data failed.", e);
//        }
//    }
//
//    protected static void parseOneRow(RowsLogEvent event, RowsLogBuffer buffer, BitSet cols, boolean isAfter)
//            throws UnsupportedEncodingException
//    {
//        TableMapLogEvent map = event.getTable();
//        if (map == null)
//        {
//            throw new RuntimeException("not found TableMap with tid=" + event.getTableId());
//        }
//
//        final int columnCnt = map.getColumnCnt();
//        final ColumnInfo[] columnInfo = map.getColumnInfo();
//
//        for (int i = 0; i < columnCnt; i++)
//        {
//            if (!cols.get(i))
//            {
//                continue;
//            }
//
//            ColumnInfo info = columnInfo[i];
//            buffer.nextValue(info.type, info.meta);
//
//            if (buffer.isNull())
//            {
//                //
//            }
//            else
//            {
//                final Serializable value = buffer.getValue();
//                if (value instanceof byte[])
//                {
//                    System.out.println(new String((byte[]) value));
//                }
//                else
//                {
//                    System.out.println(value);
//                }
//            }
//        }
//
//    }
//}
//
